package com.education.realtime.app.dwd.log;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.education.realtime.app.BaseAppV1;
import com.education.realtime.common.Constant;
import com.education.realtime.util.AtguiguUtil;
import com.education.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.HashMap;

public class DwdBaseLog extends BaseAppV1 {
    public static void main(String[] args) {
        new DwdBaseLog().init(3005,"DwdBaseLog", Constant.TOPIC_ODS_LOG);
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        // 1. etl过滤
        SingleOutputStreamOperator<String> etlStream = etl(stream);
        // 2. 新老客户纠正
        SingleOutputStreamOperator<JSONObject> validateNewStream = validateNewOrOld(etlStream);
        // 3. 分流
        HashMap<String, DataStream<String>> splitStream = split(validateNewStream);
        // 4. 不同的流写入到不同的topic
        WriteToKafkaTopic(splitStream);
    }

    private void WriteToKafkaTopic(HashMap<String, DataStream<String>> splitStream) {
        splitStream.get("action").addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_ACTION));
        splitStream.get("display").addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_DISPLAY));
        splitStream.get("error").addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_ERR));
        splitStream.get("start").addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_START));
        splitStream.get("page").addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_PAGE));
        splitStream.get("app").addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_APPVIDEO));

    }


    //json字符串是json对象的string形式，也就是string，但是符合json对象的格式 若是json对象，那就可以使用getXxx(..)来得到某key对应的value了
    private SingleOutputStreamOperator<JSONObject> validateNewOrOld(SingleOutputStreamOperator<String> etlStream) {
        return etlStream
                .map(JSON::parseObject)
                .keyBy(jsonObject -> jsonObject.getJSONObject("common").getString("mid"))
                .process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    private ValueState<String> firstVisitDateState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        firstVisitDateState = getRuntimeContext().getState(new ValueStateDescriptor<String>("firstVisitDateState", String.class));
                    }

                    @Override
                    public void processElement(JSONObject value,
                                               Context ctx,
                                               Collector<JSONObject> out) throws Exception {
                        Long ts = value.getLong("ts");
                        String date = AtguiguUtil.toDate(ts, "yyyy-MM-dd");
                        JSONObject common = value.getJSONObject("common");
                        String isNew = common.getString("is_new");

                        String firstvalue = firstVisitDateState.value();

                        if ("1".equals(firstvalue)) {
                            if (firstvalue == null) {
                                firstVisitDateState.update(date);
                            } else {
                                if (!date.equals(firstvalue)) {
                                    common.put("is_new", "0");
                                }
                            }
                        } else if (firstvalue == null) {
                            //是0，以前没跟新上去的
                            firstVisitDateState.update(AtguiguUtil.toDate(ts - 24 * 60 * 60 * 1000));
                        }
                        out.collect(value);
                    }
                });
    }
    //
    private HashMap<String, DataStream<String>> split(SingleOutputStreamOperator<JSONObject> valiNewStream) {

        OutputTag<String> errTag = new OutputTag<String>("error") {};
        OutputTag<String> pageTag = new OutputTag<String>("page") {};
        OutputTag<String> displayTag = new OutputTag<String>("display") {};
        OutputTag<String> actionTag = new OutputTag<String>("action") {};
        OutputTag<String> appVedioTag = new OutputTag<String>("app") {};

        SingleOutputStreamOperator<String> process = valiNewStream
                .process(new ProcessFunction<JSONObject, String>() {
                    @Override
                    public void processElement(JSONObject value,
                                               Context ctx,
                                               Collector<String> out) throws Exception {
                        // 1. 错误日志
                        JSONObject errObject = value.getJSONObject("err");

                        if (errObject != null) {
                            ctx.output(errTag, value.toJSONString());//把jsonobject换成json字符串，因为如果是jsonobject的话，他是引用类型，后面修改，全都会变化

                            value.remove("err");
                        }

                        // 2.启动
                        JSONObject start = value.getJSONObject("start");
                        if (start != null) {
                            out.collect(value.toJSONString());
                        } else {
                            //其他日志
                            JSONArray displays = value.getJSONArray("displays");
                            JSONObject common = value.getJSONObject("common");
                            Long ts = value.getLong("ts");

                            //曝光日志
                            if (displays != null) {
                                for (int i = 0; i < displays.size(); i++) {
                                    JSONObject display = displays.getJSONObject(i);
                                    display.put("common", common);
                                    display.put("ts", ts);
                                    ctx.output(displayTag, display.toJSONString());
                                }
                                value.remove("displays");
                            }

                            //动作日志
                            JSONArray actions = value.getJSONArray("actions");
                            if (actions != null) {
                                for (int i = 0; i < actions.size(); i++) {
                                    JSONObject action = actions.getJSONObject(i);
                                    action.put("common", common);
                                    action.put("ts", ts);
                                    ctx.output(actionTag, action.toJSONString());
                                }
                                value.remove("actions");
                            }

                            //页面浏览日志
                            JSONObject appVideo = value.getJSONObject("appVideo");
                            if (appVideo != null) {
                                appVideo.put("common", common);
                                appVideo.put("ts", ts);
                                ctx.output(appVedioTag, appVideo.toJSONString());
                            }
                            value.remove("appVideo");

                            //页面日志
                            JSONObject jsonObject = new JSONObject();
                            JSONObject page = value.getJSONObject("page");
                            if (page != null) {
                                jsonObject.put("common", common);
                                jsonObject.put("page", page);
                                jsonObject.put("ts", ts);
                                ctx.output(pageTag, jsonObject.toJSONString());
                            }
                        }
                    }
                });
        DataStream<String> page = process.getSideOutput(pageTag);
        DataStream<String> error = process.getSideOutput(errTag);
        DataStream<String> display = process.getSideOutput(displayTag);
        DataStream<String> action = process.getSideOutput(actionTag);
        DataStream<String> appVedio = process.getSideOutput(appVedioTag);

        HashMap<String, DataStream<String>> map = new HashMap<>();
        map.put("start", process);
        map.put("page", page);
        map.put("display",display);
        map.put("error",error);
        map.put("action",action);
        map.put("app",appVedio);

        return map;
    }



    private SingleOutputStreamOperator<String> etl(DataStreamSource<String> stream) {
        return stream.filter(s -> {
            try {
                JSON.parseObject(s);
                return true;
            } catch (Exception e) {
                System.out.println("json格式不对，不要瞎吉儿乱输入");
                return false;
            }
        });
    }
}
